/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.data625.commons;

import cn.ac.iie.ulss.data625.config.Configuration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

/**
 *
 * @author AlexMu
 */
public class RuntimeEnv {

    private static Configuration conf = null;
    public static final String JETTY_SERVER_IP = "jettyServerIP";
    public static final String JETTY_SERVER_PORT = "jettyServerPort";
    public static final String JETTY_SERVER_THREAD_POOL_SIZE = "jettyServerThreadPoolSize";

    public static final String ZK_CLUSTER = "zkCluster";
    public static final String ZK_LOADCLUSTER_PATH = "zkCluster.path";

    public static final String FILTER_IPS = "filterIPs";
    public static final String PACKAGE_SIZE_RESTRAINT = "packeSizeRestraint";

    public static final String DEFAULT_REGION = "defaultRegion";

    public static final String FLOW_CONTROL_ENABLE = "flowControlEnable";

    public static final String FILE_NUM = "fileNum";
    public static final String FILE_NAME_COMMON = "fileNameCommon";
    public static final String FILE_PATH = "filePath";

    private static int preLoadClusterSize = 0;
    public static boolean loadClusterSizeChanged = false;
    private static List<String> loadClusterList = new ArrayList<String>();

    public static ZkClient zc = null;
    private static Map<String, Object> dynamicParams = new HashMap<String, Object>();

    //logger
    static Logger logger = null;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = Logger.getLogger(RuntimeEnv.class.getName());
    }

    public static boolean initialize(Configuration pConf) throws Exception {

        if (pConf == null) {
            logger.error("configuration object is null");
            return false;
        }
        conf = pConf;

        getConfAndSaveAsString(JETTY_SERVER_IP, false);
        getConfAndSaveAsInteger(JETTY_SERVER_PORT);
        getConfAndSaveAsInteger(JETTY_SERVER_THREAD_POOL_SIZE);

        getConfAndSaveAsString(FILTER_IPS, false);
        getConfAndSaveAsInteger(PACKAGE_SIZE_RESTRAINT);

        getConfAndSaveAsString(DEFAULT_REGION, false);

        getConfAndSaveAsBoolean(FLOW_CONTROL_ENABLE);

        getConfAndSaveAsInteger(FILE_NUM);
        getConfAndSaveAsString(FILE_NAME_COMMON, false);
        getConfAndSaveAsString(FILE_PATH, false);

        if (!initZK()) {
            return false;
        }

        return true;
    }

    public static void addParam(String pParamName, Object pValue) {
        synchronized (dynamicParams) {
            dynamicParams.put(pParamName, pValue);
        }
    }

    public static Object getParam(String pParamName) {
        return dynamicParams.get(pParamName);
    }

    public static String getParamAsString(String pParamName) {
        return (String) dynamicParams.get(pParamName);
    }

    public static Integer getParamAsInteger(String pParamName) {
        return (Integer) dynamicParams.get(pParamName);
    }

    public static Boolean getParamAsBoolean(String pParamName) {
        return (Boolean) dynamicParams.get(pParamName);
    }

    public static void getConfAndSaveAsInteger(String confName) throws Exception {
        String tmp = conf.getString(confName, "");
        if (tmp.isEmpty()) {
            throw new Exception("parameter " + confName + " does not exist or is not defined");
        }
        try {
            dynamicParams.put(confName, Integer.parseInt(tmp));
        } catch (Exception ex) {
            throw new Exception("parameter " + confName + " 's format is wrong for " + ex.getMessage(), ex);
        }
    }

    public static void getConfAndSaveAsString(String confName, boolean allowEmpty) throws Exception {
        String tmp = conf.getString(confName, "");
        if (tmp.isEmpty() && !allowEmpty) {
            throw new Exception("parameter" + confName + "does not exist or is not defined");
        }
        dynamicParams.put(confName, tmp);
    }

    public static void getConfAndSaveAsBoolean(String confName) throws Exception {
        String tmp = conf.getString(confName, "");
        if (tmp.isEmpty()) {
            throw new Exception("parameter " + confName + " does not exist or is not defined");
        }
        try {
            dynamicParams.put(confName, Boolean.parseBoolean(tmp));
        } catch (Exception ex) {
            throw new Exception("parameter " + confName + " 's format is wrong for " + ex.getMessage(), ex);
        }
    }

    public static boolean initZK() {
        //Initialize Zookeeper
        String zkConnectionString = conf.getString(ZK_CLUSTER, "");
        if (zkConnectionString.isEmpty()) {
            logger.error("parameter " + ZK_CLUSTER + " does not exist or is not defined");
            return false;
        }
        dynamicParams.put(ZK_CLUSTER, zkConnectionString);
        try {
            zc = new ZkClient(zkConnectionString);
        } catch (Exception ex) {
            logger.error("Zookeeper client connection failed for " + ex);
        }

        try {
            if (!zc.exists("/ulss")) {
                zc.createPersistent("/ulss");
            }
            if (!zc.exists("/ulss/dd")) {
                zc.createPersistent("/ulss/dd");
            }
            if (!zc.exists("/ulss/dd/loadcluster")) {
                zc.createPersistent("/ulss/dd/loadcluster");
            }
            zc.createEphemeral("/ulss/dd/loadcluster/" + getParamAsString(RuntimeEnv.JETTY_SERVER_IP) + ":" + getParamAsInteger(RuntimeEnv.JETTY_SERVER_PORT), "");
            loadClusterList = zc.getChildren("/ulss/dd/loadcluster");
            preLoadClusterSize = loadClusterList.size();
            zc.subscribeChildChanges("/ulss/dd/loadcluster", new LoadClusterWatcher());
            logger.debug("subcribe child changes for load cluster successfully.");
        } catch (Exception ex) {
            logger.error("Cann't register localhost to zk for ", ex);
        }
        String zkLoadclusterPath = conf.getString(ZK_LOADCLUSTER_PATH, "");
        if (zkLoadclusterPath.isEmpty()) {
            logger.error("parameter [dd.zookeeper.loadcluster.path] does not exist or is not defined");
            return false;
        }
        dynamicParams.put(ZK_LOADCLUSTER_PATH, zkLoadclusterPath);
        try {
            if (!zc.exists(zkLoadclusterPath)) {
                zc.createPersistent(zkLoadclusterPath, true);
            }
        } catch (Exception ex) {
            logger.error("Zookeeper client connection failed for " + ex);
        }

        if (loadClusterList.isEmpty()) {
            logger.error("no dd server is running.");
            return false;
        }
        Timer checker = new Timer("loadcluster-checker");
        checker.schedule(new LoadClusterChecker(), 0, 30000);
        return true;
    }

    public static void dumpEnvironment() {
        conf.dumpConfiguration();
    }

    public static List<String> getLoadcluster() {

        String postfix = "/dataload/?region=" + getParamAsString(DEFAULT_REGION) + "#";
        for (int i = 0; i < loadClusterList.size(); i++) {
            String node = loadClusterList.get(i);
            if (!node.endsWith(postfix)) {
                loadClusterList.set(i, node + postfix);
            }
        }

        return loadClusterList;
    }

    public static int getLoadClusterSize() {
        return loadClusterList.size();
    }

    private static class LoadClusterWatcher implements IZkChildListener {

        @Override
        public void handleChildChange(String string, List<String> list) throws Exception {

            loadClusterList = list;
            logger.info("Load CLuster node list updated to " + list);
            if (loadClusterList.size() != preLoadClusterSize) {
                loadClusterSizeChanged = true;
                preLoadClusterSize = loadClusterList.size();
            }
        }

    }

    private static class LoadClusterChecker extends TimerTask {

        String local = getParamAsString(JETTY_SERVER_IP) + ":" + getParamAsInteger(JETTY_SERVER_PORT);
        String localFull = getParamAsString(RuntimeEnv.JETTY_SERVER_IP) + ":" + getParamAsInteger(RuntimeEnv.JETTY_SERVER_PORT) + "/dataload/?region=" + getParamAsString(DEFAULT_REGION) + "#";

        @Override
        public void run() {
            try {
                if (!zc.exists("/ulss")) {
                    zc.createPersistent("/ulss");
                }
                if (!zc.exists("/ulss/dd")) {
                    zc.createPersistent("/ulss/dd");
                }
                if (!zc.exists("/ulss/dd/loadcluster")) {
                    zc.createPersistent("/ulss/dd/loadcluster");
                }
                loadClusterList = zc.getChildren("/ulss/dd/loadcluster");
                if (!loadClusterList.contains(local)) {
                    zc.createEphemeral("/ulss/dd/loadcluster/" + local, "");
                    loadClusterList.add(localFull);
                    logger.info("re-register local node to zk successfully.");
                }
                if (loadClusterList.size() != preLoadClusterSize) {
                    loadClusterSizeChanged = true;
                    preLoadClusterSize = loadClusterList.size();
                    logger.info("load cluster size updated to " + preLoadClusterSize);
                }
            } catch (Exception ex) {
                logger.error("load cluster checker create ephemeral node failed for " + ex.getMessage());
            }
        }

    }
}
